package com.cn.broker;

import com.cn.common.MessageQueue;

import java.util.HashSet;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 锁定节点
 */
public class RebalanceLockManager {

    //已经锁定的队列
    private ConcurrentMap<String /*group*/ , ConcurrentMap<MessageQueue, LockClientEntity>> lockNodeTables = new ConcurrentHashMap<>();
    private final ReentrantLock reentrantLock = new ReentrantLock();

    public void unLock(String clientId, String group) {
        try {
            this.reentrantLock.lockInterruptibly();
            try {
                Iterator<ConcurrentMap.Entry<MessageQueue, LockClientEntity>> iterator = this.lockNodeTables.get(group).entrySet().iterator();
                while (iterator.hasNext()) {
                    ConcurrentMap.Entry<MessageQueue, LockClientEntity> entry = iterator.next();
                    if (entry.getValue().getClientId().equals(clientId)) {
                        iterator.remove();
                    }
                }
            } finally {
                this.reentrantLock.unlock();
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    /**
     * 锁定
     *
     * @param queueSet
     * @param clientId
     * @param group
     * @return
     */
    public Set<MessageQueue> lockQueue(Set<MessageQueue> queueSet, String clientId, String group) {
//        System.out.println("要锁定的长度：" + queueSet.size());
        //存放已锁定
        Set<MessageQueue> lockmq = new HashSet<>(queueSet.size());
        //存放未锁定
        Set<MessageQueue> unLockmq = new HashSet<>(queueSet.size());

        for (MessageQueue messageQueue : queueSet) {
            if (tryLock(messageQueue, group, clientId)) {
                //锁定成功
                lockmq.add(messageQueue);
            } else {
                //没有锁定成功
                unLockmq.add(messageQueue);
            }
        }

        if (!unLockmq.isEmpty()) {
            try {
                this.reentrantLock.lockInterruptibly();
                try {
                    for (MessageQueue messageQueue : unLockmq) {
                        ConcurrentMap<MessageQueue, LockClientEntity> groupLockEntity = this.lockNodeTables.get(group);
                        if (groupLockEntity == null) {
                            groupLockEntity = new ConcurrentHashMap<>();
                            this.lockNodeTables.put(group, groupLockEntity);
                        }

                        //查询该队列是否已经存在
                        LockClientEntity clientEntity = groupLockEntity.get(messageQueue);
                        if (clientEntity == null) {
                            clientEntity = new LockClientEntity();
                            clientEntity.setClientId(clientId);
                            groupLockEntity.put(messageQueue, clientEntity);
                        }

                        //尝试锁定该队列
                        if (tryLock(messageQueue, group, clientId)) {
//                            System.out.println(messageQueue.toString() + "-->锁定成功！");
                            lockmq.add(messageQueue);
                            continue;
                        }

                        LockClientEntity old = groupLockEntity.get(messageQueue);
                        if (old != null) {
                            //过期了，失去联系（注：消费端每隔一段时间都会重新锁定一次，因此假设长时间没有联系更新，那么下次操作的时候，更新为新的clietnId）
                            if (old.isExpired()) {
                                System.out.println(String.format("%s该队列旧节点[%s]已经过期，重置新节点[%s]", messageQueue.toString(), old.getClientId(), clientId));
                                old.setClientId(clientId);
                                old.setLastUpdateTime(System.currentTimeMillis());
                                lockmq.add(messageQueue);
                                continue;
                            }
                        }

//                        System.out.println(messageQueue + "==>锁定失败，节点：" + clientId);
                    }
                } catch (Exception e) {

                } finally {
                    this.reentrantLock.unlock();
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

        }
        return lockmq;
    }

    public boolean tryLock(MessageQueue messageQueue, String group, String clientId) {
        ConcurrentMap<MessageQueue, LockClientEntity> groupLockEntity = this.lockNodeTables.get(group);
        if (groupLockEntity != null) {
            LockClientEntity lockClientEntity = groupLockEntity.get(messageQueue);
            if (lockClientEntity != null) {
                if (clientId.equals(lockClientEntity.clientId)) {
                    lockClientEntity.setLastUpdateTime(System.currentTimeMillis());
                    return true;
                }
            }
        }
        return false;
    }


    class LockClientEntity {

        private String clientId;
        private long lastUpdateTime = System.currentTimeMillis();

        public boolean isLock(String clientId) {
            boolean lock = clientId.equals(this.clientId);
            return lock && !isExpired();
        }

        public boolean isExpired() {
            return (System.currentTimeMillis() - this.lastUpdateTime) > 1000 * 60;
        }

        public String getClientId() {
            return clientId;
        }

        public void setClientId(String clientId) {
            this.clientId = clientId;
        }

        public long getLastUpdateTime() {
            return lastUpdateTime;
        }

        public void setLastUpdateTime(long lastUpdateTime) {
            this.lastUpdateTime = lastUpdateTime;
        }

        @Override
        public String toString() {
            return "LockClientEntity{" +
                    "clientId='" + clientId + '\'' +
                    ", lastUpdateTime=" + lastUpdateTime +
                    '}';
        }
    }
}
